package com.rykj.energy.task.message.mq;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.*;
import com.rykj.energy.common.utils.CollectionUtils;
import com.rykj.energy.common.utils.SysConfig;
import com.rykj.energy.database.service.IExecutorService;
import com.rykj.energy.task.message.service.IMsgService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
//@Lazy
public class CustomerClient {

	public static int custMqPort = Integer.parseInt(SysConfig.getValue("custom_mq_port", "5672"));
	public static String custMqHostName = SysConfig.getValue("custom_mq_ip", "127.0.0.1");
	public static String custMqUser = SysConfig.getValue("custom_mq_user", "ruiyan");
	public static String custMqPwd = SysConfig.getValue("custom_mq_pwd", "ruiyan");
    public static String custRoutingKey_msg = SysConfig.getValue("custom_mq_routingKeyMsg", "routing_msg");
	public static String custQueue_msg = SysConfig.getValue("custom_mq_queueMsg", "msg");
    @Autowired
	private IMsgService msgService;


	@Autowired
	IExecutorService executorService;

	Connection conn = null;

	//缓存计算端最大的id
	Map<String,Integer> executorIdMap=new HashMap<>();

	@PostConstruct
	public void init(){
		try {

			ConnectionFactory factory = new ConnectionFactory();
			factory.setHost(custMqHostName);
			factory.setPort(custMqPort);
			factory.setUsername(custMqUser);
			factory.setPassword(custMqPwd);
			conn = factory.newConnection();
			receiveMessage(custQueue_msg, custRoutingKey_msg);
			List<Map<String,Object>> executors =executorService.getExecutorsDescId(null);

			if(CollectionUtils.isNotEmpty(executors)){
				executorIdMap.put("executorId",Integer.valueOf(executors.get(0).get("id")+""));
				for (Map<String,Object> exeMap:executors) {
					receiveMessage(exeMap.get("ipPort")+"", custRoutingKey_msg);
				}
			}


		}catch (Exception e){
			log.error("mq认证异常",e);
		}
	}
		
	public void receiveMessage(String queue, String routingKey) throws Exception {
		Channel channel = conn.createChannel();
		if(channel ==null) {
			init();
			channel = conn.createChannel();
		}
		channel.queueDeclare(queue, true, false, false, null);
		Channel finalChannel = channel;
		channel.basicConsume(queue, false, routingKey, new DefaultConsumer(finalChannel) {
				@Override
				public void handleDelivery(String consumerTag,
										   Envelope envelope,
										   AMQP.BasicProperties properties,
										   byte[] body){
					long deliveryTag = 0l;
					try {
						deliveryTag = envelope.getDeliveryTag();
						msgService.dealTaskMsg(new String(body, "UTF-8"));
					}catch (Exception e){
						log.error("mq消费异常",e);
					}catch (Error er){
						log.error("mq消费错误",er);
					}
                    try {
						finalChannel.basicAck(deliveryTag, false);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }

                }
			});
	}
}
